package com.rem.flink.flink3Transform;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 数据转换
 * max()：在输入流上，对指定的字段求最大值。 只保留第一个字段的数据 只有比较的字段更新
 * maxBy()：与 max()类似，在输入流上针对指定字段求最大值 但会保留全部字段
 *
 * @author Rem
 * @date 2022-09-28
 */

public class TransformKeyByTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Bob", "./neek", 3000L),
                new Event("Jeck", "./ceel", 4000L),
                new Event("Linda", "./cart", 5000L),
                new Event("Bob", "./home", 6000L),
                new Event("Bob", "./cart", 2200L),
                new Event("Bob", "./index", 2300L),
                new Event("jesson", "./cart", 2400L),
                new Event("kid", "./cart", 500L)
        );
        //按照user分组  max 聚合操作后 只取聚合字段更新 其他字段选择第一条字段
        stream.keyBy((KeySelector<Event, String>) Event::getUser)
                .max("timestamp")
                .print("max:");

        //按照user分组  maxBy 聚合操作后 取所有字段
        stream.keyBy((KeySelector<Event, String>) Event::getUser)
                .maxBy("timestamp")
                .print("maxBy");


        env.execute();

    }
}
